/*
 * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package demo.parallelcore;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * A demo shows how does Stream's flatMap works. Demo shows tabulate stream 
 * element to single element stream or empty stream
 * @author tyan
 */
public class Tabulate extends DemoHelper {
    /*
     * Join supplier and customer that in same country. note not all supplier 
     * has customer
     * @return  a list of customer-supplier pair which in same counrey
    */
    public static List<Pair<Customer, Supplier>> sameCountryCustomerAndSuppliers() {
        //A BiFunction which convert one given customer, one given supplier
        //into a (customer, supplier) pair
        BiFunction<Customer, Supplier, Pair<Customer, Supplier>> combine 
                = Pair<Customer, Supplier>::new;
        Map<String,List<Customer>> countryToCustomers  = customers.
                parallelStream().
                //grouping customers by their counrty
                collect(Collectors.groupingBy(Customer::getCountry));
        //Function that get supplier by given (customer, supplier) pair 
        Function<Pair<Customer, Supplier>,String> csToSupplierName 
                = pair -> pair.getValue1().getSupplierName();
        //Function that get customer by given (customer, supplier) pair 
        Function<Pair<Customer, Supplier>,String> csToCustomerName 
                = pair -> pair.getValue0().getCompanyName();
        return suppliers.parallelStream().
                //convert every supplier and customer which in the same country
                //to a pair stream of (customer, supplier)
                flatMap(
                    //get a customers's stream, those customer is same country
                    //as given supplier
                    supplier -> countryToCustomers.getOrDefault(
                        supplier.getCountry(), new ArrayList<>()).stream().
                    map(customer -> combine.apply(customer, supplier))).
                //convert it to a List
                collect(Collectors.toList());
    }
    
    /*
     * flatMap can generate empty stream, sometimes it's helpful for ruling
     * out data
     * @return   a string set who is either customer and supplier both.
    */
    public static Set<String> bothCustomerAndSupplier() {
        Set<String> suppliersName = suppliers.parallelStream().
                //convert supplier to its name
                map(Supplier::getSupplierName).
                //get a supplier set
                collect(Collectors.toSet());
        return customers.parallelStream().
                //map customer to its name
                map(Customer::getCompanyName).
                //query supplier set with customer's name, if exist, create 
                //a stream with single element customer name, if doesn't, create
                //an empty stream
            flatMap(customer -> {
                if(suppliersName.contains(customer)) 
                    return Stream.of(customer);
                else
                    return Stream.empty(); }).
                //convert all match name into a set
                collect(Collectors.toSet());
    }
}
